package com.ljc.listener;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.google.gson.Gson;
import com.ljc.constant.RabbitMQConstant;
import com.ljc.constant.RedisConstant;
import com.ljc.entity.UserFollowing;
import com.ljc.entity.UserMoment;
import com.ljc.service.UserFollowingService;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;

@Slf4j
@Component
public class MovementListener {
    @Resource
    private StringRedisTemplate stringRedisTemplate;
    @Resource
    private UserFollowingService userFollowingService;

    /**
     * 监听发送动态普通队列
     */
    @RabbitListener(queues = RabbitMQConstant.MOVEMENT_QUEUE_NAME)
    public void sendMovementListener(String movementJson, Message message, Channel channel) throws IOException {
        String messageId = message.getMessageProperties().getMessageId();
        int retryCount = Integer.parseInt((String) Objects.requireNonNull(stringRedisTemplate.opsForHash().get(RedisConstant.MOVEMENT_MESSAGE_PREFIX + messageId, "retryCount")));
        if (retryCount > 3) {
            //重试次数大于3，直接放到死信队列
            log.error("短信消息重试超过3次：{}", messageId);
            //basicReject方法拒绝deliveryTag对应的消息，第二个参数是否requeue，true则重新入队列，否则丢弃或者进入死信队列。
            //该方法reject后，该消费者还是会消费到该条被reject的消息。
            channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
            stringRedisTemplate.delete(RedisConstant.MOVEMENT_MESSAGE_PREFIX + messageId);
            return;
        }
        UserMoment userMoment = new Gson().fromJson(movementJson, UserMoment.class);
        try {
            if (userMoment == null) {
                throw new RuntimeException("sendMovementListener参数为空");
            }
            Long userId = userMoment.getUserId();
            // 获取发布动态的用户的粉丝列表
            List<UserFollowing> fanList = userFollowingService.getUserFans(userId);
            // 将动态存入每个粉丝的空间中
            for (UserFollowing fan : fanList) {
                String key = RedisConstant.SUBSCRIBED_MOVEMENT_PREFIX + fan.getUserId();
                // 从redis中获取当前用户的动态列表
                String subscribedListStr = stringRedisTemplate.opsForValue().get(key);
                List<UserMoment> subscribedList;
                // 如果之前的动态列表为空，则新建列表
                if (StringUtils.isEmpty(subscribedListStr)) {
                    subscribedList = new ArrayList<>();
                } else {
                    // 否则将之前的动态列表数据转换成动态列表
                    subscribedList = JSONArray.parseArray(subscribedListStr, UserMoment.class);
                }
                // 将新动态加入动态列表
                subscribedList.add(userMoment);
                stringRedisTemplate.opsForValue().set(key, JSONObject.toJSONString(subscribedList));
            }
            //手动确认消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            log.info("动态发送成功：{}", userMoment);
            stringRedisTemplate.delete(RedisConstant.MOVEMENT_MESSAGE_PREFIX + messageId);
        } catch (Exception e) {
            stringRedisTemplate.opsForHash().increment(RedisConstant.MOVEMENT_MESSAGE_PREFIX + messageId, "retryCount", retryCount);
            channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
        }
    }

    /**
     * 监听到发送短信死信队列
     */
    @RabbitListener(queues = RabbitMQConstant.MOVEMENT_DELAY_QUEUE_NAME)
    public void movementDelayQueueListener(String movementJson, Message message, Channel channel) throws IOException {
        UserMoment userMoment = new Gson().fromJson(movementJson, UserMoment.class);
        try {
            log.error("监听到死信队列消息==>{}", userMoment);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception e) {
            channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
        }
    }
}
